/*
 * Middleware
 * Copyright (C) 2010/11
 * Riccardo Pulvirenti, Giuseppe Ravidà, Andrea Tino
 *
 * This program is free software: you can redistribute it and/or modify
 * it under the terms of the GNU General Public License as published by
 * the Free Software Foundation, either version 3 of the License, or
 * (at your option) any later version.
 *
 * This program is distributed in the hope that it will be useful,
 * but WITHOUT ANY WARRANTY; without even the implied warranty of
 * MERCHANDABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
 * GNU General Public License for more details.
 *
 * You should have received a copy of the GNU General Public License
 * along with this program. If not, see http://www.gnu.org/licenses
 */

/*!
 * File: manager.cpp
 */

#include <iostream>
#include <string>
#include <exception>
#include <boost/thread.hpp>
#include <boost/asio.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/archive/binary_iarchive.hpp>
#include <boost/archive/binary_oarchive.hpp>
#include <cstdlib>

#include "manager.hpp"
#include "tqueue_proxy_factory.hpp"
#include "task_proxy_factory.hpp"
#include "data_proxy_factory.hpp"
#include "discovery_proxy_factory.hpp"
#include "task_descriptor.hpp"
#include "utils.hpp"
#include "address.hpp"

using namespace middleware;

//-----------------------------------------------------------------------------
// Constructors and destructor
/* Default constructor */
Manager::Manager() {
	this->_config.man_ip_toman_bind = "127.0.0.1"; /* Not used to bind */
	this->_config.man_port_toman_bind = this->createPort();
	this->_config.ptrto_worker = 0;
	this->_config.ptrto_discovery = 0;
	// Logger
	std::string logfile = std::string("logfile_man_");
	this->_logger.set_log_filename(logfile);
	this->_logger.set_log_tag("MAN");
}
/* Configuration constructor */
Manager::Manager(ManagerConfig conf) {
	this->_config.man_ip_toman_bind =
			conf.man_ip_toman_bind; /* Not used to bind */
	this->_config.man_port_toman_bind = conf.man_port_toman_bind;
	this->_config.ptrto_worker= conf.ptrto_worker;
	this->_config.ptrto_discovery = conf.ptrto_discovery;
	// Logger
	std::string logfile = std::string("logfile_man_") + conf.log_postfix;
	this->_logger.set_log_filename(logfile);
	this->_logger.set_log_tag("MAN");
}
/* Destructor */
Manager::~Manager() {
	// Every member is statically allocated.
}
//-----------------------------------------------------------------------------
// Private members
/* Gets a random port */
std::string Manager::createPort() {
	return utils::GetRandomIntegralNumberStr(kManagerMinManPort,
			kManagerMaxManPort);
}
/* Thread execution flow | Sharing a SynQueue */
void Manager::exec() {
	// Execution flow for Manager
	std::cout << kManagerEtq << "Running!" << std::endl;
	this->bonjour(); /* Initialization sequence */
	try {
		// Launching thread to manage incoming tasks
		std::cout << kManagerEtq << "Launching task manager thread..." <<
				std::endl;
		this->_taskmanager_thread = boost::shared_ptr<boost::thread>(
				new boost::thread(boost::bind(&Manager::exec_taskmanager,
						this)));
	} catch(std::exception& e) {
		std::cout << kManagerEtq <<
				"Error in task manager thread launch: " <<
				e.what() << std::endl;
		std::cout << kManagerEtq << "Now exiting!" << std::endl;
		this->_manager_thread.reset();
		return;
	}
	try {
		// Cycling and dequeueing from queue
		for (;;) {
			std::cout << kManagerEtq << "Dequeueing loop, " <<
					"Dequeueing workflow..." << std::endl;
			Workflow wf = this->_wfqueue_manager.dequeue(); /* Blocking call */
			std::cout << kManagerEtq << "Dequeued workflow id=" <<
					wf.get_id() << " (" << wf.size() <<
					" tasks). WFQueue has now " <<
					this->_wfqueue_manager.size() << " workflows." <<
					" Now managing dequeued wf..." << std::endl;
			this->_logger.Print(std::string("Dequeued workflow WID=") +
					boost::lexical_cast<std::string>(wf.get_id()) +
					std::string(" (") +
					boost::lexical_cast<std::string>(wf.size()) +
					std::string(" tasks)."));
			// A thread will manage the workflow (multithread)
			boost::shared_ptr<boost::thread>(new boost::thread(
					boost::bind(&Manager::ManageWorkflow, this, wf)));
			//this->ManageWorkflow(wf); /* Managing this flow (singlethread) */
		}
	} catch(std::exception& e) {
		std::cout << kManagerEtq << "Error: " << e.what() << std::endl;
		std::cout << kManagerEtq << "Now exiting!" << std::endl;
		return;
	}
}
/* Thread execution flow for task manager for incoming tasks */
void Manager::exec_taskmanager() {
	std::cout << kManagerEtq << kManager1Etq << "Now running!" << std::endl;
	/* Listen for incoming tasks, from other peers, and redirect them to the
	 * worker: using two proxies: for man_to_man and for man_to_worker */
	try {
		std::cout << kManagerEtq << kManager1Etq <<
				"Creating and configuring task proxy factory..." << std::endl;
		TaskProxyFactory task_proxy_factory(kReceiveTaskProxy,
				this->_config.man_ip_toman_bind,
				this->_config.man_port_toman_bind);
		std::cout << kManagerEtq << kManager1Etq <<
				"Task proxy factory configured: bind_ip=" <<
				this->_config.man_ip_toman_bind << ", bind_port=" <<
				this->_config.man_port_toman_bind<< std::endl;
		std::cout << kManagerEtq << kManager1Etq <<
				"Creating and configuring tqueue proxy factory..." << std::endl;
		TQueueProxyFactory tqueue_proxy_factory(this->_config.ptrto_worker);
		std::cout << kManagerEtq << kManager1Etq <<
				"Tqueue proxy factory configured: Pointer to Worker ok" << std::endl;
		// Get proxies
		TaskManagerProxy* task_proxy = task_proxy_factory.getProxy();
		TQueueManagerProxy* tqueue_proxy = tqueue_proxy_factory.getProxy();
		// Here I have proxies too
		std::cout << kManagerEtq << kManager1Etq << "Got proxies!" << std::endl;
		std::cout << kManagerEtq << kManager1Etq <<
				"Entering task management loop..." << std::endl;
		// Managing incoming task descriptors
		for (;;) {
			std::cout << kManagerEtq << kManager1Etq <<
					"Waiting for incoming task..." << std::endl;
			TaskDescriptor td = task_proxy->recv_task(); /* Blocking */
			std::cout << kManagerEtq << kManager1Etq << "TASK ARRIVED: TID=" <<
					td.get_id() << ". Now enqueueing..." << std::endl;
			// Sending to worker (enqueueing)
			tqueue_proxy->enqueue(td); /* Locking call */
			std::cout << kManagerEtq << kManager1Etq <<
					"Task TID=" << td.get_id() <<
					" finally enqueued in worker!" << std::endl;
			this->_logger.Print(std::string("Arrived task from net TID=") +
					boost::lexical_cast<std::string>(td.get_id()) +
					std::string(" and enqueued in worker."));
		}
	} catch(std::exception& e) {
		std::cout << kManagerEtq << kManager1Etq <<	"Error: " <<
				e.what() << std::endl;
		std::cout << kManagerEtq << kManager1Etq <<
				"Program now dying..." << std::endl;
	}
}
/* Called through proxy by ui to enqueue a wf | Sharing a SynQueue */
void Manager::enqueuer(const Workflow& wf) {
	std::cout << kManagerEtq << "Workflow enqueuer called!" << std::endl;
	try {
		std::cout << kManagerEtq << "Workflow to enqueue: WID=" <<
				wf.get_id() << " (" << wf.size() <<
				" tasks). Now enqueueing..."  << std::endl;
		// Now enqueueing the workflow
		Workflow wfcopy = wf;
		this->_wfqueue_manager.enqueue(wfcopy);
		std::cout << kManagerEtq <<	"Workflow " << wf.get_id() <<
				" enqueued!" << std::endl;
		this->_logger.Print(std::string("Enqueued workflow WID=") +
				boost::lexical_cast<std::string>(wf.get_id()) +
				std::string(" (") +
				boost::lexical_cast<std::string>(wf.size()) +
				std::string(" tasks)"));
	} catch(std::exception& e) {
		std::cout << kManagerEtq <<	"Error: " << e.what() << std::endl;
		std::cout << kManagerEtq <<	"Program now dying..." << std::endl;
		exit(1);
	}
}
/* Workflow manager (can be a thread), sending tasks */
void Manager::ManageWorkflow(Workflow workflow) {
	std::cout << kManagerEtq << "WFManager: " << "Managing workflow WID="
			<< workflow.get_id() << " (" << workflow.size() << ")" <<
			std::endl;
	try {
		// Getting proxy
		DiscoveryProxyFactory proxy_factory(this->_config.ptrto_discovery);
		WorkerDiscoveryProxy* proxy = proxy_factory.getProxy();
		Workflow::TaskCollectionConstIterator it;
		// Do for every task in the workflow
		for (it = workflow.begin(); it < workflow.end(); it++) {
			std::cout << kManagerEtq <<	"Managing task: TID=" <<
					it->get_id() << " of WID=" << workflow.get_id() <<
					std::endl;
			Address addr = proxy->get_worker(*it);
			std::cout << kManagerEtq << "Task TID=" << it->get_id() <<
					" of WID=" << workflow.get_id() <<
					" has been associated to worker at address IP=" <<
					addr.get_ip() << " and PORT=" << addr.get_port() <<
					" (man)" << std::endl;
			// I got address to send this task to, let's send it
			TaskProxyFactory task_proxy_factory(kSendTaskProxy,
					addr.get_ip(),
					boost::lexical_cast<std::string>(addr.get_port()));
			TaskManagerProxy* task_proxy = task_proxy_factory.getProxy();
			bool error = false;
			try {
				task_proxy->send_task(*it);
				std::cout << kManagerEtq <<	"TASK SENT: " <<
						"Task TID=" << it->get_id() <<
						" of WID=" << workflow.get_id() <<
						" was successfully sent to manager of peer at IP=" <<
						addr.get_ip() << " and PORT=" << addr.get_port() <<
						" (man)" << std::endl;
				this->_logger.Print(std::string("Task sent to net TID=") +
						boost::lexical_cast<std::string>(it->get_id()) +
						std::string(", sent to machine IP=") +
						addr.get_ip() + std::string(" and PORT=") +
						boost::lexical_cast<std::string>(addr.get_port()) +
						std::string("."));
			} catch(ProxyException& ex) {
				std::cout << kManagerEtq <<	"Task sending error: " <<
						"Unable to send task TID=" << it->get_id() <<
						" of WID=" << workflow.get_id() <<
						". Task will be aborted! " << ex.what() << std::endl;
				this->_logger.Print(
						std::string("Unable to send to net task TID=") +
						boost::lexical_cast<std::string>(it->get_id()) +
						std::string(", to machine IP=") +
						addr.get_ip() + std::string(" and PORT=") +
						boost::lexical_cast<std::string>(addr.get_port()) +
						std::string(" (man)."));
				error = true;
			}
			/* Now sending data, if this is an executable task, to worker
			 * of the destination peer, ATTENTION, worker may not find
			 * data because here it must still be sent, well, worker
			 * has reasonable logic to consider this possibility */
			if ((it->get_type() == kTaskExecutable) && (!error)) {
				std::cout << kManagerEtq <<	"Sending data for task TID=" <<
						it->get_id() << " (executable)" << std::endl;
				try {
					DataProxyFactory data_proxy_factory(kSendDataProxy,
							addr.get_ip(),
							boost::lexical_cast<std::string>(addr.get_port_w()));
					DataManagerProxy* data_proxy =
							data_proxy_factory.getProxy();
					// I have proxy
					Data data("AUTO DATA", 1001);
					data.set_owner_task_id(it->get_id());
					data_proxy->send_data(data);
					std::cout << kManagerEtq <<	"TASK DATA SENT: " <<
							"Data for task TID=" << it->get_id() <<
							" of WID=" << workflow.get_id() <<
							" was successfully sent to worker of peer at IP=" <<
							addr.get_ip() << " and PORT=" << addr.get_port_w() <<
							" (w)" << std::endl;
					this->_logger.Print(
							std::string("Sent executable task data (PUSH): TID=") +
							boost::lexical_cast<std::string>(it->get_id()) +
							std::string(", to machine IP=") +
							addr.get_ip() + std::string(" and PORT=") +
							boost::lexical_cast<std::string>(addr.get_port_w()) +
							std::string(" (w). Data was: ") +
							data.get_data_str() + std::string("/") +
							boost::lexical_cast<std::string>(data.get_data_int()) +
							std::string("."));
				} catch(ProxyException& ex) {
					std::cout << kManagerEtq <<	"Task data sending error: " <<
							"Unable to send data for task TID=" << it->get_id() <<
							" of WID=" << workflow.get_id() <<
							". Data will be aborted! " << ex.what() << std::endl;
					this->_logger.Print(
							std::string("Could not send executable task data (PUSH): TID=") +
							boost::lexical_cast<std::string>(it->get_id()) +
							std::string(", to machine IP=") +
							addr.get_ip() + std::string(" and PORT=") +
							boost::lexical_cast<std::string>(addr.get_port_w()) +
							std::string(" (w). "));
				}
			}
		}
	} catch(ProxyException& e) {
		std::cout << kManagerEtq <<	"Error: " << e.what() << std::endl;
	} catch(std::exception& e) {
		std::cout << kManagerEtq <<	"Error: " << e.what() << std::endl;
	}
}
/* Bonjour */
void Manager::bonjour() {
	std::cout << kManagerEtq << "Initialization sequence now started..." <<
				std::endl;
	std::cout << kManagerEtq <<
			"Configuration loaded: TaskArrivalGate_IP: this=" <<
			this->_config.man_ip_toman_bind << ", TaskArrivalGate_PORT: " <<
			this->_config.man_port_toman_bind << "; Pointer to Worker ok" <<
			std::endl;
	std::cout << kManagerEtq << "Initialization completed!" << std::endl;
}
